package cc.verywell.pureblock.net;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.util.EncryptUtil;

/**
* 
*
* @author TanYaqiu 
* @address flat 601
* @date 2019年2月23日 下午3:41:04
*/
public class AioServer implements Closeable {
	static private final boolean DEBUG = true;
	
	final private AioProvider provider;
	public AioServer(AioProvider provider) {
		this.provider = provider;
	}
	
	private ArrayList<AsynchronousServerSocketChannel> assList = new ArrayList<>();
	public synchronized SocketAddress start(AsynchronousChannelGroup acg,InetSocketAddress bind) throws IOException {
		for (AsynchronousServerSocketChannel ass : assList) {
			if(bind.equals(ass.getLocalAddress()))throw new IOException("server exist!");
		}
		if(acg==null) acg = provider.getDefaultChannelGroup(0);
		//acg = AsynchronousChannelGroup.withThreadPool(Executors.newSingleThreadExecutor());
		AsynchronousServerSocketChannel ass = AsynchronousServerSocketChannel.open(acg);
		//ass.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
		ass.bind(bind,AioProvider.MAX_PENDING_CONNECTIONS);
		assList.add(ass);
		///如果退出时发现ShutdownChannelGroupException 那是因为  connection accept after group has shutdown , 查了一下是java问题，可以忽略.
		ass.accept(ass, acceptHandler);
		timeoutRunner();
		return ass.getLocalAddress();
	}
	
	private CompletionHandler<AsynchronousSocketChannel,AsynchronousServerSocketChannel> acceptHandler = new CompletionHandler<AsynchronousSocketChannel,AsynchronousServerSocketChannel>() {
	      public void completed(AsynchronousSocketChannel ch, AsynchronousServerSocketChannel ass) {
//	    	  if(DEBUG)Pblog.print("DEBUG","accept:",ch);
	    	  //先注册下一个请求
	    	  ass.accept(ass,this);
	    	  if(provider.isTooManyPendingConnections()){
	    		  try {
	  					ch.close();
	    		  } catch (IOException e) {
	  					if(Pblog.IO)Pblog.IO(e);
	    		  }
		    	  return;
	    	  }
	    	  provider.pendingSet.add(ch);
				try {
					ch.setOption(StandardSocketOptions.SO_REUSEADDR, true);
					ch.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
					HandshakeChannelServer hc = handshakeChannels.poll();
					if(hc==null)hc = new HandshakeChannelServer(provider.myAlias()); //这里会throw NoSuchAlgorithmException
					hc.init(ch, AioProvider.TIME_OUT, acceptHandshake);
				} catch (Throwable e) {
					if(Pblog.IO)Pblog.IO(e);
					try {
						ch.close();
					} catch (IOException e1) {
						if(Pblog.IO)Pblog.IO(e);
					}
					provider.pendingSet.remove(ch);
					//e.printStackTrace();
					return;
				}

	    	  
	      }
	      public void failed(Throwable exc, AsynchronousServerSocketChannel ass) {
				if(Pblog.IO && ass!=null && ass.isOpen())Pblog.IO(exc);
	      }
	  };

	
	private BlockingQueue<HandshakeChannelServer> handshakeChannels = new LinkedBlockingQueue<>(AioProvider.MAX_HANDSHAKE_CHANNEL_CACHE);
	private HandshakeFunction<HandshakeChannelServer> acceptHandshake = new HandshakeFunction<HandshakeChannelServer>() {
		@Override
		public void handshakeCompleted(HandshakeChannelServer obj, EncryptUtil eu, ByteBuffer unread, int salt) {
			try {
				NormalChannel nc = provider.handshakeCompleted(obj.getChannel(), eu, unread, null, salt, false);
				if(nc==null) {
					// 致命错误
				}
			}finally {
				handshakeChannels.offer(obj.clearUser());
			}
		}
		@Override
		public void handshakeFailed(HandshakeChannelServer obj, Throwable exp) {
			try {
				provider.handshakeFaild(obj.getChannel(), exp);
			}finally {
				handshakeChannels.offer(obj.clearUser());
			}
		}
	};
	
	

  private ScheduledExecutorService timeoutRunner;
  private synchronized void timeoutRunner() throws IOException {
		if(timeoutRunner!=null)return;//throw new IOException("server exist!");
		int timeout = AioProvider.TIME_OUT/2;
		if(timeout<=0)return;
		timeoutRunner = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
//			private AtomicInteger count = new AtomicInteger();
			@Override
			public Thread newThread(Runnable r) {
				Thread t = new Thread(provider.THREADGROUP,r, "timeoutRunner");
				t.setDaemon(true);
				t.setPriority(t.getPriority()+1); //这边如果没信息过去，那边就要自动断线了。
				return t;
			}
		});
		timeoutRunner.scheduleWithFixedDelay(() -> {
          try {
              synchronized (timeoutRunner) {
                  if (provider.connectSet.isEmpty()) return;

                  List<NormalSession> timeOutSession = new ArrayList<NormalSession>(provider.connectSet.size());
                  
                  synchronized (provider.connectSet) {// Must be manually synched by user!
                	  long writeTime = System.currentTimeMillis()-AioProvider.TIME_OUT*1000/2;
                	  for(NormalChannel nc : provider.connectSet) {
                		  NormalSession ns = nc.getSession();
                		  if(ns.getTimelog().getLastWriteTime()<writeTime)timeOutSession.add(ns);
                	  }
                 	}
                  
                  if (timeOutSession.isEmpty()) return;

                  long readTime = System.currentTimeMillis()-AioProvider.TIME_OUT*1000;
                  for (NormalSession ns : timeOutSession) {
                	  if(ns.getTimelog().getLastReadTime()<readTime) { // 即使context那边暂停读包，这边时间够了也会自动断线;
                		  if(Pblog.IO)Pblog.IO("timeoutRunner r",readTime+AioProvider.TIME_OUT*1000-ns.getTimelog().getLastReadTime(),">",AioProvider.TIME_OUT*1000," to close ",ns);
                		  ns.close();
                	  }else {
                		  if(Pblog.IO)Pblog.IO("timeoutRunner w",readTime+AioProvider.TIME_OUT*1000-ns.getTimelog().getLastWriteTime(),">",AioProvider.TIME_OUT*1000/2," to timeout ",ns);
                		  ns.timeout();
                	  }
					
                  }
 
              }

              //Tasks.forEach(Runnable::run);
          } catch (Throwable t) {
          	if(Pblog.IO)Pblog.IO(t);
          }
      }, 0, timeout, TimeUnit.SECONDS);
  }
	
  	public synchronized boolean close(InetSocketAddress binded) {
  		Iterator<AsynchronousServerSocketChannel> it = assList.iterator();
  		while(it.hasNext()){
  			AsynchronousServerSocketChannel ass = it.next();
			try {
				if(binded.equals(ass.getLocalAddress())) {
					if(assList.size()<=1) {
						close();
					}else{
						it.remove();
						ass.close();
					}
					return true;
				}
			} catch (IOException e) {
				if(Pblog.IO)Pblog.IO(e);
			}
		}
  		return false;
  	}
	@Override
	public synchronized void close() {
		try{
			for (AsynchronousServerSocketChannel ass : assList) {
				try {
					ass.close();
				} catch (IOException e) {
					if(Pblog.IO)Pblog.IO(e);
				}
			}
		}finally {assList.clear();}
		if(DEBUG)Pblog.print("DEBUG","server ass closed");
		if(timeoutRunner!=null)try{
			timeoutRunner.shutdown();
			try {
				timeoutRunner.awaitTermination(5, TimeUnit.SECONDS);
			} catch (InterruptedException e) {
				if(Pblog.IO)Pblog.IO(e);
			}
		}finally {timeoutRunner=null;}
	}
	
	
	
	
	

}
